package cn.doitedu.demo1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/11/18
 * @Desc: 学大数据，上多易教育
 * <p>
 * -- 营销策略：营销规则
 * 针对年龄在 20~28 之间，性别为 male的用户，
 * 当他发生一个行为：添加购物车 时，即给他推送一个满200-50的优惠活动信息
 * <p>
 * 需要的数据：
 * 用户画像
 * 实时行为
 **/
public class Demo1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 读kafka中行为数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setGroupId("doit42-rtmk")
                .setClientIdPrefix("doit42-client-")
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(logJson -> JSON.parseObject(logJson, EventBean.class));


        KeyedStream<EventBean, Long> keyedStream = beanStream.keyBy(bean -> bean.getUser_id());


        SingleOutputStreamOperator<String> messages = keyedStream.process(new KeyedProcessFunction<Long, EventBean, String>() {

            RestHighLevelClient client;
            SearchRequest request;
            JSONObject message;

            @Override
            public void open(Configuration parameters) throws Exception {
                // es的请求客户端
                client = new RestHighLevelClient(
                        RestClient
                                .builder(new HttpHost("doitedu", 9200, "http"))
                );
                // 用于查询参数封装的对象
                request = new SearchRequest("doit42_userprofile");


                message = new JSONObject();
                message.put("rule_id", "rule-001");


            }

            @Override
            public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context ctx, Collector<String> out) throws Exception {

                // 判断当前行为事件，是否满足营销规则中的事件 ：add_cart
                if ("add_cart".equals(eventBean.getEvent_id())) {

                    // 判断当前行为人，是否满足营销规则中的画像条件
                    IdsQueryBuilder idQuery = QueryBuilders.idsQuery().addIds(eventBean.getUser_id() + "");
                    RangeQueryBuilder ageQuery = QueryBuilders.rangeQuery("tag0202").gte(36).lte(40);
                    MatchQueryBuilder genderQuery = QueryBuilders.matchQuery("tag0201", "male");

                    BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
                    boolQueryBuilder.must(idQuery)
                            .must(ageQuery)
                            .must(genderQuery);


                    request.source(new SearchSourceBuilder().query(boolQueryBuilder));

                    // 用客户端发送请求
                    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
                    if (response.getHits().getHits().length > 0) {
                        // 如果满足，则发送消息
                        message.put("user_id", eventBean.getUser_id());
                        message.put("hit_time", eventBean.getAction_time());
                        out.collect(message.toJSONString());
                    }
                }
            }
        });


        messages.print();

        env.execute();

    }


}
